package com.atguigu.gmall.realtime.function;


import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.util.JdbcUtil;
import com.atguigu.gmall.realtime.util.RedisUtil;
import com.atguigu.gmall.realtime.util.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author liuyun
 * @title: DimAsyncFunction
 * @projectName gmal0924
 * @description: TODO
 * @date 2022-3-22 12:54
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T, T> {

    public abstract void addDim(Jedis redisClient,
                                Connection conn,
                                T input,
                                ResultFuture<T> resultFuture) throws Exception;

    private ThreadPoolExecutor threadPool;

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        //每天一天数据就使用启动一个线程池
        threadPool.submit(new Runnable() {



            @SneakyThrows
            @Override
            public void run() {
                Connection conn = null;
                Jedis redisClient = null;
                try {
                    conn = JdbcUtil.getJdbcConnect(Constant.PHOENIX_DRIVER, Constant.PHOENIX_URL, null, null);
                    //出来维度信息
                    redisClient = RedisUtil.getRedisClient();
                    addDim(redisClient, conn, input, resultFuture);
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    // 关闭连接对象
                    if (conn != null) {
                        try {
                            conn.close();
                        } catch (SQLException e) {
                            e.printStackTrace();
                        }
                    }

                    if (redisClient != null) {
                        redisClient.close();
                    }
                }

            }
        });
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPool = ThreadPoolUtil.getThreadPool();
    }
}
